fix: match Spark's duplicate field and field id semantics in parquet field lookup - #5654
dwsmith1983 wants to merge 24 commits into
Conversation
sunchao
left a comment
There was a problem hiding this comment.
This fixes stray-name matching and placeholder collisions, and adds nested duplicate-ID rejection and last-wins exact-name lookup. One gap remains: metadata-only struct relabeling bypasses the duplicate-ID check, as detailed inline.
I compared the code with maintained Spark 3.5 and 4.0 sources. Eight component-check groups passed using extracted Comet helpers with Arrow/Parquet 58.4.0 and DataFusion 54.1.0. A separate probe reproduced the cast bypass and verified a renamed-child control. These probes use limited scaffolding and are not full Comet scan, JNI or Spark query tests. The reported 246 native and 58 Spark 3.5 tests are the author's results.
At 04:51 UTC, current-head CI had 29 successful, 32 running and 7 skipped checks. Full CI validation was still pending.
| // Mirror Spark's `foundDuplicateFieldInFieldIdLookupModeError` | ||
| // (`_LEGACY_ERROR_TEMP_2094`): a requested ID resolving to more | ||
| // than one file field is ambiguous. | ||
| Some(indices) => { |
There was a problem hiding this comment.
[P2] Run duplicate-ID validation before metadata-only struct relabeling
Could you route metadata-only struct adaptations through this validation too? For file struct s<x: int id=1, y: int id=1, z: int id=2> and requested s<x: int id=1, y: int id=3, z: int id=2>, Spark rejects requested ID 1 as ambiguous. DataFusion emits a struct cast, but CometCastColumnExpr::evaluate takes types_differ_only_in_field_names and calls relabel_array, because that predicate ignores field-ID metadata. The new lookup never runs and leaves all three physical values in place. A focused probe using the current cast expression and a real Arrow/Parquet round trip returned [42, 43, 44], while renaming requested x made the same input reach the duplicate-ID error. Could you guard the relabel shortcut for ID-based reads and add a cast-expression or scan regression with unchanged child names?
There was a problem hiding this comment.
Good catch, the shortcut sailed right past the new validation. Fixed in 3d68f22: the relabel arm is now guarded so that when use_field_id is set and the requested type carries field id metadata, evaluation falls through to the struct conversion where the duplicate id lookup runs. Chose the guard at the call site rather than inside types_differ_only_in_field_names since that predicate is a pure structural comparison with no access to the parquet options. Your exact probe is now a regression test (unchanged child names, duplicate id 1, asserts the 2094 error) plus a companion pinning that the fast path survives for name only differences without ids and for the flag alone.
3d68f22 to
e1d9eb3
Compare
e1d9eb3 to
58e67fe
Compare
|
Reviewed head
The strongest design improvement is to resolve and validate requested fields once per file schema, then reuse the mapping across batches. That addresses the validation bypasses and repeated lookup work together. Metadata-only relabeling remains safe when the resolved mapping is positional. A small mapping object is a useful abstraction here. Validation: 100 native Parquet tests passed, with default HDFS features disabled. Additional head/base probes confirmed both correctness cases. Performance evidence measures component allocations, not overall scan speed. CI snapshot: 57 passed, 7 running, 7 skipped. Nothing was posted to GitHub. |
|
Thanks, all three are addressed in a69b5f5, following the once-per-file design you suggested. The physical expression adapter factory already runs once per file schema, so it now resolves a small Your repros: the identical One residual worth naming: DataFusion's opener skips the adapter entirely when the logical and physical schemas compare equal and no predicate exists. Spark-written files always carry key-value metadata that arrow-rs folds into the physical schema, so they always go through the adapter, but a file with no metadata at all and duplicated ids inside a struct would still read positionally. Happy to cover that in a follow-up if you think it matters. |
96aa08d to
cc2ffa1
Compare
4dba460 to
13c4afc
Compare
|
@sunchao the once-per-file mapping round covering your three findings is pushed. Ready for another look. |
andygrove
left a comment
There was a problem hiding this comment.
convert_struct in native/core/src/parquet/parquet_support.rs around line 563 now calls array.column(from_index) with an index resolved at planning time against adapted_physical_schema, where the old code derived it from the runtime array's own DataType. The only guard is sources.len() != to_fields.len(), which checks the target side. What guarantees the struct array the reader hands back always carries the same children in the same order as the physical field the mapping was resolved against? If that can drift at all, this is an index panic in the executor rather than a DataFusionError. Checking from_index against array.num_columns() would bound the worst case.
The description lists last-wins exact-name resolution as one of the three fixes and resolve_struct_mapping does it for struct children. At the top level in case-sensitive mode with no field ids, needs_remap is false in schema_adapter.rs around line 520, so resolution falls to DefaultPhysicalExprAdapter, which goes through Schema::index_of and returns the first match. Spark builds caseSensitiveParquetFieldMap at the root message level with the same .toMap it uses for nested groups. Was the top level deliberately left out of scope?
The field-id ambiguity path is covered from several angles now. The case-insensitive name ambiguity that resolve_struct_mapping raises around line 388 does not appear to have a companion test in the new struct_field_matching module. It might be worth pinning that half too, since it is the branch that decides between an error and a silently wrong column.
On the residual you named where DataFusion skips the adapter when the two schemas compare equal and there is no predicate, I confirmed that short circuit in the 55.0.0 opener. Could you open a tracking issue and link it here so it does not get lost? The branch also conflicts with main right now and needs a rebase before anything meaningful runs against it.
|
Rebased onto main and the three points are in the head (b438dce). Bounds: Root last-wins: not deliberate, the top level had simply fallen to the default adapter. With duplicate exact names at the root in case-sensitive mode the remap path now runs, the shadowed earlier fields get a placeholder name so the default adapter's The case-insensitive ambiguity now has its companion test in The opener short circuit is tracked in #5801. |
…tead of picking the last
The field-id gating gap is #5936, linked from both PRs. On behavior, this PR now does what Andy proposed on #5786 for the nested case rather than last-wins: a requested field that matches more than one byte-identical sibling is refused with a message naming the field, while reading the unique sibling beside them still works, both pinned in Rust and by a Scala read of a file whose struct carries On order: since the refusal now lives in the resolver this PR owns, I would land this one first and rebase #5786 onto it for the shapes that remain, the metadata-time check included if it is still needed. The fallible name folding from #5845 is already propagated here. |
|
Reviewed head P2: Footer validation rejects supported duplicate-root readsparquet_support.rs:352 treats the root schema as a nested struct, applying the new duplicate-name rejection before the adapter can select the first root column. Trigger: a file contains
Disabling field-ID reads makes the PR return the correct rows. The unrelated ID-bearing column activates the footer check and causes the failure. Fix: preserve root first-wins behavior in footer validation while retaining nested rejection. Add a mixed ID/name regression through the reader factory. Validation
Full review and evidence (local review report and reproduction logs). Nothing posted to GitHub. |
…ing requested nested ones
Done. |
|
Agreed on landing this first, then integrating the remaining nested-only checks in #5786. The separate mixed-type root investigation in #5964 now has parquet-java PR apache/parquet-java#3796. Spark descriptor/filter changes have passed targeted experimental checks, but still need upstream JIRA tracking, a current-master port and an available parquet dependency. That work remains separate from these Comet PRs. |
comphead
left a comment
There was a problem hiding this comment.
Let me try to come up with more light weight solution
|
Triage note: #5786 is working the other side of duplicate Parquet field names in the native scan. It rejects ambiguous sibling names before the decoder is built, closing #5783, and explicitly leaves "Spark-compatible duplicate-name resolution" as separate work — which is this PR. You share The risk is that the two disagree on the outcome for the same file: this PR resolves case-sensitive duplicates last-wins the way Spark's |
|
@dwsmith1983 yours have a lot more work implemented. I see no problem in landing it first and I adapt mine after. @andygrove when do you intent to close 1.1.0? Just want to know the feasibility to have both on this release |
There is no hard deadline, but I was hoping to have 1.1.0 released before the end of the month. |
…eader tests Main now declines any requested schema that repeats a Parquet field id at planning time, so the two tests that asked for both id-1 fields no longer reached the native scan. Each now requests only one of them, keeps the scan native, and still expects Spark's duplicate field error from the reader.
Order as ErikBPF said above: this PR first, #5786 adapts after. On the split: #6004 now declines at planning any requested schema that repeats an id, so I have dropped the equal-schema rationale from the tests here and they request a single id that the file carries twice. What this PR still adds is native raising Spark's error when a requested id is ambiguous in the file, last-wins exact-name lookup like Spark's
That check is what #6004 landed for field ids, and it stays. The reader-side error covers the case a plan-level check cannot see, an id the requested schema names once that the file carries twice. If you would rather keep one mechanism, I can trim the footer check to that file-side case; the adapter path has to stay for the Spark error text. |
|
Thanks @dwsmith1983 @ErikBPF, folks, WDYT to proceed with #5786 to patch physical scans and then check what cases are still not addressed in this field and fix pursue them in this PR? The initial reason I was slightly concerned is 2k LOC to patch an extreme scenario |
We can cut it down. The footer check's reason to exist went away with #6004, so I can drop it and its tests, which takes out the reader-factory and the Rust test bulk. What stays is the adapter raising Spark's error for an ambiguous requested id, last-wins exact names, the id shield and typed errors through the JNI cause chain, none of which #5786 covers. On order, #5786 currently does not type-check against main per @sunchao and would reject the DataFusion-written file you showed on #6004, so I would still land this first and let #5786 rebase onto the smaller diff. Does that work? |
Which issue does this PR close?
Closes #5801.
Part of the restructuring of #5365 requested in review: this extracts the duplicate field and field id matching semantics that previously traveled with the Delta contrib work, re-derived on top of the folding that #5602 added. It also absorbs the reader-factory validation that was stacked on it as #5808, since that commit cannot rebase onto main on its own and the two could not merge separately.
Rationale for this change
Four places where the native parquet field lookup diverges from Spark:
remap_physical_schemaonly shields id-bearing logical fields whose id is missing from the file. When a logical field's id matches one physical field but a stray physical column carries that logical field's name, the stray column can still name-match through the expression adapter fallback and hijack the read.parquet_convert_struct_to_structsilently resolves a requested field id that matches more than one physical field to the first match. Spark raises the duplicate field error in field id lookup mode..toMap, where the last entry wins.What changes are included in this PR?
SparkError::DuplicateFieldByFieldId(_LEGACY_ERROR_TEMP_2094). Duplicate ids that no requested field references remain harmless, at the root as well as inside nested types: the adapter records a root ambiguity per logical field and raises it fromrewriteonly for the columns a read references, the same way nested ambiguities are handled. Before this, a duplicate id on columns the read never asked for failed the query, while Spark answers it becauseclipParquetSchemaonly sees the requested schema.List<Struct>read as aLargeList<Struct>(or any other pair of list representations) resolves its element fields with Spark's rules before Arrow changes the list layout. This is the same shape fix: apply Spark's Parquet conversion rules to nested struct/list/map fields #5681 takes for the converter, so whichever of the two lands second rebases mechanically.SparkErrorreachable through aContextorSharedwrapper anywhere in the plan now keeps its class. An ANSI failure in a pushed-down filter is unaffected, because DataFusion stringifies it into anArrowError::ComputeErrorand nothing typed survives the chain.Both validation paths now validate the same scope, the columns a read requests. The footer check covers the required schema, and the adapter raises only for referenced columns. The check also catches a duplicate id on root fields and inside a
list<struct>element, so it closes the same opener-skip gap for the root-level check, not only the nested one.There is no equivalent gap for the case-insensitive duplicate name error: for the file and requested schemas to compare equal, the requested schema would have to hold two fields folding to the same name, and Spark rejects that at analysis with
COLUMN_ALREADY_EXISTS. Two fields can share an id while having distinct names, which is what makes the duplicate id error reachable and theuse_field_id && schema_holds_field_idsgate on the footer check sufficient.How are these changes tested?
Rust, all written before the change they pin and failing on the previous head unless noted:
rewriteof an unreferenced column succeeds, of the ambiguous one errors).List<Struct>requested asLargeList<Struct>resolves to a non-positional element mapping, converts by id, and null-fills a requested element field the file lacks instead of reading its neighbour.DataSourceExecon a file written without key-value metadata, asserting first that the file schema equals the requested schema so the opener skips the adapter: duplicate struct id rejected, unique ids read, check inert with field id matching off, the same through the planner's data schema plus projection wiring, duplicate id at the root, duplicate id inside alist<struct>element, case-insensitive duplicate name (_LEGACY_ERROR_TEMP_2093) rejected at footer load with the case-sensitive read of the same file succeeding, and unrequested root duplicate ids read through the adapter path.SparkErrorbehind DataFusion'sContextandExternalwrappers; a plain Parquet error still classifies as a file read failure.Core crate 367 tests, bridge crate 32, clippy with
-D warningsand fmt clean. Each commit compiles and passes the parquet module tests on its own.Scala: a
ParquetReadV1Suitecase writes the file with parquet-mr and no key-value metadata, asserts the footer's key-value map is empty and that the plan carries the native scan, and expects Spark's duplicate field id error; it reported no exception against the previous native library.ParquetReadV1Suite,CometNativeReaderSuiteandSparkErrorConverterSuiteon Spark 3.5 against the rebuilt native library: 137 succeeded, 0 failed.